// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-11-19

package kafka_plugin

import (
    "gitee.com/go-libs/log/common"
    "gitee.com/go-libs/log/conf"
    "github.com/IBM/sarama"
    "time"
)

// 输出日志.
type writer struct {
    err      error
    producer sarama.SyncProducer
}

// +------------------------------------------------------------------------------+
// | Interface methods                                                            |
// +------------------------------------------------------------------------------+

func (o *writer) Write(formatter common.LoggingFormatter, items []common.Item) (err error) {
    var (
        cfg      = conf.New()
        messages = make([]*sarama.ProducerMessage, 0)
    )

    // 1. 内部错误.
    //    例如建立 Kafka 连接出错.
    if err = o.err; err != nil {
        return
    }

    // 2. 构建消息.
    for _, item := range items {
        messages = append(messages, &sarama.ProducerMessage{
            Topic:     cfg.KafkaAdapter.Topic,
            Value:     sarama.StringEncoder(formatter.ToString(item)),
            Timestamp: item.Time(),
        })
    }

    // 3. 发送消息.
    err = o.producer.SendMessages(messages)
    return
}

// +------------------------------------------------------------------------------+
// | Access methods                                                               |
// +------------------------------------------------------------------------------+

func (o *writer) init() *writer {
    var (
        c  = conf.New()
        kc = sarama.NewConfig()
    )

    kc.Net.DialTimeout = time.Duration(c.KafkaAdapter.DialTimeout) * time.Second
    kc.Net.ReadTimeout = time.Duration(c.KafkaAdapter.ReadTimeout) * time.Second
    kc.Net.WriteTimeout = time.Duration(c.KafkaAdapter.WriteTimeout) * time.Second

    kc.Producer.RequiredAcks = sarama.NoResponse
    kc.Producer.Timeout = time.Duration(c.KafkaAdapter.WriteTimeout) * time.Second
    kc.Producer.Retry.Max = 1
    kc.Producer.Retry.Backoff = 300 * time.Millisecond
    kc.Producer.Return.Errors = true
    kc.Producer.Return.Successes = true
    kc.Producer.CompressionLevel = sarama.CompressionLevelDefault

    o.producer, o.err = sarama.NewSyncProducer(c.KafkaAdapter.Addresses, kc)
    return o
}
